package com.tool.engine.export.engine;

import com.tool.engine.export.exception.ExportTaskException;
import com.tool.engine.export.mertics.Collector;
import com.tool.engine.export.output.DataOutputSource;
import com.tool.engine.export.processor.ExportTaskProcessor;
import com.tool.engine.export.task.Task;
import com.tool.engine.export.task.TaskDecorator;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;

@Slf4j
@Setter
public abstract class AbstractExportTaskEngine implements ExportTaskEngine {

    private final ExecutorService defaultPool =  new ForkJoinPool();

    private Collector collector;

    private TaskDecorator taskDecorator;

    @Override
    public final Future<?> processTask(Task<?> task, ExportTaskProcessor processor, DataOutputSource dataOutPutSource, ExecutorService threadPoolExecutor) throws ExportTaskException {
        try {
            collector.incrExportTaskCount();
            if (threadPoolExecutor == null) {
                return defaultPool.submit(createRunnable(task, processor, dataOutPutSource));
            }
            return threadPoolExecutor.submit(createRunnable(task, processor, dataOutPutSource));
        }catch (Exception e) {
            collector.decrExportTaskCount();
            throw new ExportTaskException(e.getMessage());
        }
    }

    @Override
    public final Future<?> processTask(Task<?> task, ExportTaskProcessor processor, DataOutputSource dataOutPutSource) throws ExportTaskException {
        return this.processTask(task, processor, dataOutPutSource, null);
    }

    private Runnable createRunnable(Task<?> task, ExportTaskProcessor processor, DataOutputSource dataOutPutSource) {
        return taskDecorator.decorate(() -> {
            long startTime = System.currentTimeMillis();
            try {
                collector.incrProcessTask();
                processor.processTask(task, dataOutPutSource);
                if (processor.isCompleted()) {
                    log.info("process task success name:{}", task.getFileName());
                    this.handlerSuccess(task);
                } else {
                    log.warn("process task failed name:{}", task.getFileName());
                    this.handlerFailed(task);
                }
            } catch (Exception e) {
                log.warn("process task name:{} occur exception:{}", task.getFileName(),e.getMessage());
                handlerException(e, task);
            } finally {
                dataOutPutSource.complete();
                long costTime = System.currentTimeMillis()- startTime;
                log.info("process task completed name:{}, costTime:{}ms", task.getFileName(), costTime);
                collector.decrProcessTask();
                collector.decrExportTaskCount();
                collector.incrTotalTime(costTime);
                if (Objects.nonNull(processor)) {
                    processor.clear();
                }
                this.handlerComplete(task);
            }
        });
    }

    public void handlerException(Exception e, Task<?> task){

    }

    public void handlerComplete(Task<?> task){

    }

    public void handlerFailed(Task<?> task){

    }

    public void handlerSuccess(Task<?> task){

    }

}
